package com.asggo.g21.utils;

import static com.asggo.g21.constant.CacheConstants.WEB_SOCKET_TOPIC;

import cn.hutool.core.collection.CollUtil;
import com.asggo.g21.dto.WebSocketMessageDTO;
import com.asggo.g21.holder.WebSocketSessionHolder;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.function.Consumer;
import lombok.extern.slf4j.Slf4j;
import org.springframework.web.socket.PongMessage;
import org.springframework.web.socket.TextMessage;
import org.springframework.web.socket.WebSocketMessage;
import org.springframework.web.socket.WebSocketSession;

/**
 * Created by IntelliJ IDEA.
 *
 * @author eric 2023/12/15 16:28
 */
@Slf4j
public class WebSocketUtils {

  private WebSocketUtils() {

  }

  /**
   * 发送消息
   *
   * @param sessionKey session主键 一般为用户id
   * @param message    消息文本
   */
  public static void sendMessage(Long sessionKey, String message) {
    WebSocketSession session = WebSocketSessionHolder.getSession(sessionKey);
    sendMessage(session, message);
  }

  /**
   * 订阅消息
   *
   * @param consumer 自定义处理
   */
  public static void subscribeMessage(Consumer<WebSocketMessageDTO> consumer) {
    RedisUtils.subscribe(WEB_SOCKET_TOPIC, WebSocketMessageDTO.class, consumer);
  }

  /**
   * 发布订阅的消息
   *
   * @param webSocketMessage 消息对象
   */
  public static void publishMessage(WebSocketMessageDTO webSocketMessage) {
    List<Long> unsentSessionKeys = new ArrayList<>();
    // 当前服务内session,直接发送消息
    for (Long sessionKey : webSocketMessage.getSessionKeys()) {
      if (WebSocketSessionHolder.existSession(sessionKey)) {
        WebSocketUtils.sendMessage(sessionKey, webSocketMessage.getMessage());
        continue;
      }
      unsentSessionKeys.add(sessionKey);
    }
    // 不在当前服务内session,发布订阅消息
    if (CollUtil.isNotEmpty(unsentSessionKeys)) {
      WebSocketMessageDTO broadcastMessage = new WebSocketMessageDTO();
      broadcastMessage.setMessage(webSocketMessage.getMessage());
      broadcastMessage.setSessionKeys(unsentSessionKeys);
      RedisUtils.publish(
          WEB_SOCKET_TOPIC,
          broadcastMessage,
          consumer ->
              log.info(" WebSocket发送主题订阅消息topic:{} session keys:{} message:{}",
                  WEB_SOCKET_TOPIC, unsentSessionKeys, webSocketMessage.getMessage()
              )
      );
    }
  }

  /**
   * 发布订阅的消息(群发)
   *
   * @param message 消息内容
   */
  public static void publishAll(String message) {
    WebSocketMessageDTO broadcastMessage = new WebSocketMessageDTO();
    broadcastMessage.setMessage(message);
    RedisUtils.publish(
        WEB_SOCKET_TOPIC,
        broadcastMessage,
        consumer
            -> log.info("WebSocket发送主题订阅消息topic:{} message:{}", WEB_SOCKET_TOPIC, message)
    );
  }

  public static void sendPongMessage(WebSocketSession session) {
    sendMessage(session, new PongMessage());
  }

  public static void sendMessage(WebSocketSession session, String message) {
    sendMessage(session, new TextMessage(message));
  }

  private static void sendMessage(WebSocketSession session, WebSocketMessage<?> message) {
    if (session == null || !session.isOpen()) {
      log.error("[send] session会话已经关闭");
    } else {
      try {
        session.sendMessage(message);
      } catch (IOException e) {
        log.error("[send] session({}) 发送消息({}) 异常", session, message, e);
      }
    }
  }

}
